package com.peng.sparktest.sparkcore

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object SparkApiTest03_Aggregator {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setAppName("pv_uv_job").setMaster("local")
    val context = new SparkContext(conf)
    context.setLogLevel("ERROR")
    val data: RDD[(String, Int)] = context.parallelize(List(
      ("xiaoming", 2),
      ("xiaoming", 3),
      ("xiaoming", 7),
      ("xiaoming", 8),
      ("xiaohong", 9),
      ("xiaohong", 6),
      ("xiaoli", 1),
      ("xiaoli", 22),
      ("xiaoli", 24)
    ))
    //    //行转列
    //    val groupRes: RDD[(String, Iterable[Int])] = data.groupByKey()
    //    groupRes.foreach(println)
    //    //列转行
    //    //以下两种作用相同：
    //
    //    println("================flatMap========================")
    //    groupRes.flatMap(map => map._2.map(v => (map._1, v))).foreach(println)
    //
    //
    //    println("================flatMapValues========================")
    //    groupRes.flatMapValues(v => v).foreach(println)
    //
    //    //转换出每个key下的前两条排好序的value的值
    //    println("================mapValues========================")
    //    groupRes.mapValues(values => values.toList.sorted.take(2)).foreach(println)
    //
    //    //如果是flatMapValues  就是扁平化处理
    //    groupRes.flatMapValues(values => values.toList.sorted.take(2)).foreach(println)

    println("================SUM========================")
    val sum: RDD[(String, Int)] = data.reduceByKey(_ + _)
    sum.foreach(println)

    println("================MAX========================")
    val max: RDD[(String, Int)] = data.reduceByKey((oldV, newV) => {
      if (oldV > newV) {
        oldV
      } else {
        newV
      }
    })
    max.foreach(println)

    println("================MAX========================")
    val min: RDD[(String, Int)] = data.reduceByKey((oldV, newV) => {
      if (oldV < newV) {
        oldV
      } else {
        newV
      }
    })
    min.foreach(println)

    println("===============COUNT========================")

    val count: RDD[(String, Int)] = data.mapValues(vals => 1).reduceByKey(_ + _)
    count.foreach(println)

    println("===============AVG========================")
    val joinRes: RDD[(String, (Int, Int))] = sum.join(count)
    val avg: RDD[(String, Int)] = joinRes.mapValues((tumple: (Int, Int)) => tumple._1 / tumple._2)
    avg.foreach(println)
    println("===============AVG-COMBINE========================")
    //我们根据SPARK的任务工作执行图发现，这样的求平均值会造成两个结果集要进行两次计算，两次shuffle，汇总到一起后，再进行计算，
    //有没有什么办法可以优化求平均值的方法， 也就是把两个结果集尝试变成一个？？？
    //有的，直接调用 combineByKey方法来实现自定义处理
    val avgCombine: RDD[(String, (Int, Int))] = data.combineByKey(
      //      createCombiner: V => C,  当第一次碰到某个key时，对其值得处理
      (e: Int) => (e, 1),
      //      mergeValue: (C, V) => C, 当已经某个key时，将已有该key的值 和 本次值传入 进行处理
      (oldV: (Int, Int), newV: Int) => (oldV._1 + newV, oldV._2 + 1),
      //      mergeCombiners: (C, C) => C) //当对溢写结果进行合并时的处理
      (v1: (Int, Int), v2: (Int, Int)) => (v1._1 + v2._1, v1._2 + v2._2)
    )
    avgCombine.mapValues(e => e._1 / e._2).foreach(println)

    while (true) {

    }
  }

}
